/*
 * Copyright 2013 Eduard Sedov.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.es.util;

import java.nio.ByteBuffer;
import org.es.io.Page;

/**
 *
 * @author Eduard Sedov
 */
public class Bucket extends Page {

    public static final int TAIL_SIZE = 8;
    public static final int KEY_LENGTH_SIZE = 2;
    private static final int BUCKET_INDEX_OFFSET = 0;
    private static final int NEXT_BUCKET_OFFSET = 4;

    protected Bucket(ByteBuffer buffer) {
        super(buffer);
    }

    public static Bucket get(ByteBuffer buffer) {
        return new Bucket(buffer);
    }

    public static Bucket create(ByteBuffer buffer) {
        return new Bucket(Page.initEmptyPage(buffer, TAIL_SIZE));
    }

    public ByteBuffer getKey(int index) {
        ByteBuffer item = getItem(index);
        int keyLength = item.getShort();
        ByteBuffer key = item.slice();
        key.limit(keyLength);
        return key;
    }

    public ByteBuffer getValue(int index) {
        ByteBuffer item = getItem(index);
        int keyLength = item.getShort();
        item.position(KEY_LENGTH_SIZE + keyLength);
        return item.slice();
    }

    public boolean hasSpace(ByteBuffer key, ByteBuffer value) {
        return hasSpace(KEY_LENGTH_SIZE + key.limit() + value.limit());
    }

    public boolean hasSpace(byte[] key, byte[] value) {
        return hasSpace(KEY_LENGTH_SIZE + key.length + value.length);
    }

    public void put(ByteBuffer key, ByteBuffer value) {
        ByteBuffer item = ByteBuffer.allocate(KEY_LENGTH_SIZE + key.limit() + value.limit());
        item.putShort((short) key.limit());
        int keyPosition = key.position();
        item.put(key);
        key.position(keyPosition);
        int valuePosition = value.position();
        item.put(value);
        value.position(valuePosition);
        int index = search(key);
        if (index < 0) {
            index = (-1 * index) - 1;
        } else {
            removeItem(index);
        }
        addItem(item, index);
    }

    public void put(byte[] key, byte[] value) {
        ByteBuffer item = ByteBuffer.wrap(new byte[KEY_LENGTH_SIZE + key.length + value.length]);
        item.putShort((short) key.length);
        item.put(key);
        item.put(value);
        int index = search(key);
        if (index < 0) {
            index = (-1 * index) - 1;
        } else {
            removeItem(index);
        }
        addItem(item, index);
    }

    public void remove(byte[] key) {
        remove(ByteBuffer.wrap(key));
    }

    public void remove(ByteBuffer key) {
        int index = search(key);
        if (index >= 0) {
            removeItem(index);
        }
    }

    /**
     * Searches the given key in this bucket.
     *
     * @param key byte buffer to be searched for
     * @return index of the search key, if it is contained in the array within
     * the specified range; otherwise, <tt>(-(<i>insertion point</i>) - 1)</tt>.
     * The <i>insertion point</i> is defined as the point at which the key would
     * be inserted into the array: the index of the first element in the range
     * greater than the key, or <tt>toIndex</tt> if all elements in the range
     * are less than the specified key. Note that this guarantees that the
     * return value will be &gt;= 0 if and only if the key is found.
     */
    public int search(byte[] key) {
        ByteBuffer searchKey = ByteBuffer.wrap(key);
        return search(searchKey);
    }

    /**
     * Searches the given key in this bucket.
     *
     * @param key byte array to be searched for
     * @return index of the search key, if it is contained in the array within
     * the specified range; otherwise, <tt>(-(<i>insertion point</i>) - 1)</tt>.
     * The <i>insertion point</i> is defined as the point at which the key would
     * be inserted into the array: the index of the first element in the range
     * greater than the key, or <tt>toIndex</tt> if all elements in the range
     * are less than the specified key. Note that this guarantees that the
     * return value will be &gt;= 0 if and only if the key is found.
     */
    public int search(ByteBuffer key) {
        int low = 0;
        int high = count() - 1;
        while (low <= high) {
            int mid = (low + high) >>> 1;
            ByteBuffer currentKey = getKey(mid);
            int compareResult = currentKey.remaining() - key.remaining();
            if (compareResult == 0) {
                compareResult = currentKey.compareTo(key);
            }
            if (compareResult < 0) {
                low = mid + 1;
            } else if (compareResult > 0) {
                high = mid - 1;
            } else {
                return mid; // key found
            }
        }
        return -(low + 1);
    }

    public int bucketIndex() {
        return tail().getInt(BUCKET_INDEX_OFFSET);
    }

    public void bucketIndex(int index) {
        tail().putInt(BUCKET_INDEX_OFFSET, index);
    }

    public int nextBucket() {
        return tail().getInt(NEXT_BUCKET_OFFSET);
    }

    public void nextBucket(int index) {
        tail().putInt(NEXT_BUCKET_OFFSET, index);
    }
}
